package com.ibm.cps.kafka.listener;

import org.apache.log4j.Logger;

import com.google.common.base.Throwables;
import com.ibm.cps.kafka.KafkaTopicConsumeThread;
import com.ibm.cps.newmessage.AbstractMetadata;
import com.ibm.cps.newmessage.MetadataFactory;
import com.ibm.util.TopologyStreamIds;
import com.ibm.util.exception.CPSException;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.tuple.Values;

public class DeleteProcessorListener extends KafkaTopicConsumeThread {

	private SpoutOutputCollector collector;
	private Logger logger;

	public DeleteProcessorListener(String zookeeper, String topicname, SpoutOutputCollector collector) {
		super(zookeeper, topicname);
		this.collector = collector;
		logger = Logger.getLogger(DeleteProcessorListener.class);
	}

	@Override
	public void process(byte[] byteMessage) throws CPSException {
		try {
			String message = new String(byteMessage);
			logger.info("Receive deleted metadata " + message);
			AbstractMetadata metadata = MetadataFactory.parseJsonMetadata(message);
			collector.emit(TopologyStreamIds.PROCESSOR_DELETE_SPOUT_STREAM, new Values(metadata));
		} catch (CPSException e) {
			logger.error(Throwables.getStackTraceAsString(e));
		}
	}
}
